package com.xl.flinkdemo.processAPI;

import com.xl.flinkdemo.entity.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.expressions.In;
import org.apache.flink.util.Collector;

/**
 * @ClassName KeyedProcessFunction
 * @Description TODO
 * @Author hxl
 * @Date 2021/5/24 17:17
 * Version 1.0
 **/
public class BiaoZhunKeyedProcessFunction {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);

    ParameterTool parameterTool = ParameterTool.fromArgs(args);
    String host = parameterTool.get("hostname");
    int port = parameterTool.getInt("port");



    // 从socket文本流读取数据
    // nc -lk 7777(linux自带的Navicat工具)
    DataStream<String> inputdataStream = env.socketTextStream(host, port);


    DataStream<SensorReading> resultStream = inputdataStream.map(line -> {
      String[] fields = line.split(",");
      return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    });

    resultStream.keyBy("id").process(new TempContinueIncreWarning(10)).print();

    env.execute();
  }

  //实现自定义的处理函数，检查一段时间内的温度连续上升，输出报警
  public static class TempContinueIncreWarning extends KeyedProcessFunction<Tuple, SensorReading, String> {

    //定义私有属性，当前统计的时间间隔区域
    private Integer interval;

    public TempContinueIncreWarning(Integer interval) {
      this.interval = interval;
    }

    //定义状态，保存上一个温度值，和定时器的时间戳
    ValueState<Long> tsTimerState;
    ValueState<Double> lastTempState;


    @Override
    public void open(Configuration parameters) throws Exception {
      tsTimerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts-timer", Long.class));
      //初始值，第一次进来是null，定为double类型的最小值
      lastTempState =
          getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class, Double.MIN_VALUE));
    }

    @Override
    public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {

      Double lastTemp = lastTempState.value();
      Long timerTs = tsTimerState.value();
      Long ts = value.getTimestamp() + interval * 1000L;
      //温度上升且没有定时器的时候，注册10秒后的定时器，开始等待
      if (value.getSensors() - lastTemp > 0) {
        if (timerTs == null) {
          ctx.timerService().registerEventTimeTimer(ts);
          tsTimerState.update(ts);
        }
      } else {
        //删除定时器
        ctx.timerService().deleteEventTimeTimer(ts);
      }
      lastTempState.update(value.getSensors());

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
      out.collect("传感器："+ctx.getCurrentKey().getField(0)+"温度值连续"+interval+"上升");
      lastTempState.clear();
    }
  }
}
